This is the consumer API for kafka.